package roles

import (
	"time"

	"gitee.com/cjianwen/go/demo/socks_v2/raft_demo/server/global"
	"gitee.com/cjianwen/go/demo/socks_v2/raft_demo/server/proto"
	"gitee.com/cjianwen/go/demo/socks_v2/raft_demo/server/types"
	"gitee.com/cjianwen/go/logs"
)

/*
raft 节点之前的通讯协议处理
*/
type protocol struct {
	*roleState

	pingPack           *proto.PackData                   // ping 包数据，重复利用
	tempDatas          map[uint32]*types.AppendEntryData // 不完整的临时数据
	lastSendApppendIdx uint32
	lastSendCommitIdx  uint32
	maxLearnIdx        uint32
	lastRecvCommitIdx  uint32 // append 完成后，这里可以自动提交，因为 leader 等待半数机器完成后就开始提交数据，commit 有可能会先于 apppend 到来
}

func (p *protocol) RequestVote() {
	pack := proto.NewDefPackWithData(proto.CM_RequestVote, nil)
	pack.Head.DataIndex = global.DataEntries.GetIndex()
	pack.Head.CommitId = global.DataEntries.GetEntryItemTerm(pack.Head.DataIndex - 1)
	pack.Head.Term = p.tryVoteTerm
	if err := global.NodeConns.SendPackToAll(pack); err != nil {
		logs.Warn("RequestVote error: %s ", err)
	}
}

func (p *protocol) SendPing(appendIdx uint32) {
	p.pingPack.Head.Term = p.Term
	p.pingPack.Head.CommitId = global.DataEntries.GetCommitIdx()
	p.pingPack.Head.DataIndex = appendIdx
	global.NodeConns.SendPackToAll(p.pingPack)
}

func (p *protocol) SendAppend(dataIdx uint32) bool {
	data := global.DataEntries.GetEntryItem(dataIdx)
	if data == nil {
		logs.Printf("SendAppend 获取数据失败: %d", dataIdx)
		return false
	}

	p.appendRespCount = 1 // 包括自己一个
	p.lastSendApppendIdx = dataIdx
	prevTerm := global.DataEntries.GetEntryItemTerm(dataIdx - 1)
	logs.Printf("SendAppend: %d, prevTerm = %d", dataIdx, prevTerm)
	appendData := types.AppendEntryData{EntryItem: *data, AppendIdx: dataIdx, PrevTerm: prevTerm}
	logs.Printf("SendAppend: appendData = %#+v", appendData)
	if bits, ok := types.Encode(&appendData); ok {
		pack := proto.NewDefPackWithData(proto.CM_Append, bits)
		pack.Head.Term = p.Term
		pack.Head.DataIndex = dataIdx
		if err := global.NodeConns.SendPackToAll(pack); err != nil {
			logs.Warn("AddData error: send is failed: %s", err)
			return false
		}
	} else {
		logs.Warn("SendAppend 数据系列化错误: ")
		return false
	}

	return true
}

func (p *protocol) SendCommit(commitId uint32) bool {
	p.commitRespCount = 1 // 自己也算一个
	p.lastSendCommitIdx = commitId
	pack := proto.NewDefPackWithData(proto.CM_Commit, nil)
	pack.Head.Term = p.Term
	pack.Head.CommitId = commitId
	pack.Head.DataIndex = global.DataEntries.GetIndex()
	if err := global.NodeConns.SendPackToAll(pack); err != nil {
		logs.Warn("SendCommit error %s", err)
		return false
	}
	return true
}

// maxLearnIdx 外面传进来的是没减1的
func (p *protocol) SendLearn(dataIdx, maxLearnIdx uint32) {
	if maxLearnIdx > 0 {
		p.maxLearnIdx = maxLearnIdx
	}

	var prevTerm uint32
	if data := global.DataEntries.GetEntryItem(dataIdx - 1); data != nil {
		prevTerm = data.Term
	}

	logs.Printf("SendLearn: term = %d, idx = %d", prevTerm, dataIdx)

	pack := proto.NewDefPackWithData(proto.CM_Learn, nil)
	pack.Head.Term = prevTerm
	pack.Head.DataIndex = dataIdx
	if err := global.LeaderConn.SendPackToQueue(pack); err != nil {
		logs.Printf("SendLearn error: %s", err)
	}
}

func (p *protocol) on_CM_Ping(conn *proto.ConnClient, pack *proto.PackData) {
	// logs.Printf("收到ping, term = %d, myTerm = %d \n", pack.Head.Term, p.role.Term)
	if pack.Head.Term > p.Term {
		if !p.IsFollower() {
			logs.Printf("on_CM_Ping 收到 leader ping,切换为 follower from %s, Term = %d ", p.GetRoleName(), pack.Head.Term)
			p.becomeFollower()
		} else {
			logs.Println("Follower 收到 leader ping, Term = ", pack.Head.Term)
		}
		p.Term = pack.Head.Term
		p.LastRecvPingTime = time.Now()
		global.LeaderConn = conn
	} else if pack.Head.Term == p.Term {
		p.LastRecvPingTime = time.Now() // 正常的也要更新时间
		if global.LeaderConn != conn {
			logs.Println("on_CM_Ping 在ping中设置 LeaderConn")
			global.LeaderConn = conn
		}
		if p.leaderPingDataIdx != pack.Head.DataIndex {
			p.leaderPingDataIdx = pack.Head.DataIndex
			logs.Printf("on_CM_Ping dataIndex = %d", pack.Head.DataIndex)
		}
	}

	pk := proto.NewDefPackWithData(proto.SM_PONG, nil)
	pk.Head.Term = p.Term
	pk.Head.DataIndex = uint32(global.ServerInfo.ServerId)
	err := conn.SendPackToQueue(pk)
	if err != nil {
		logs.Println("ProcPing error: ", err)
	}
}

func (p *protocol) on_SM_Pong(conn *proto.ConnClient, pack *proto.PackData) {
	// nothing
}

// 处理投票
func (p *protocol) on_CM_RequestVote(conn *proto.ConnClient, pack *proto.PackData) {
	var isOk uint32
	idx := global.DataEntries.GetIndex()
	// 最后一条数据的任期号要大于等于本节点最后一条数据的任期号
	if pack.Head.Term > p.Term && pack.Head.Term > p.tryVoteTerm && pack.Head.DataIndex >= idx && pack.Head.CommitId >= global.DataEntries.GetEntryItemTerm(idx-1) {
		p.tryVoteTerm = pack.Head.Term
		isOk = 1 // 投赞成票
		if !p.IsFollower() {
			logs.Printf("收到投票请求, 改变角色, 成为 follower %d -> 0, Term = %d ", p.currRole, pack.Head.Term)
			p.becomeFollower()
		}
		p.LastRecvPingTime = time.Now() // 假定是一个 ping ,不那么快竞选
	}
	logs.Printf("收到投票请求: term = %d, voteDataIdx = %d, myDataIdx = %d, result = %d ", pack.Head.Term, pack.Head.DataIndex, global.DataEntries.GetIndex(), isOk)
	pk := proto.NewDefPackWithData(proto.SM_RequestVote, nil)
	pk.Head.CommitId = isOk
	if err := conn.SendPackToQueue(pk); err != nil {
		logs.Println("procProposal error: ", err)
	}
}

// 处理投票
func (p *protocol) on_SM_RequestVote(conn *proto.ConnClient, pack *proto.PackData) {
	if p.IsCandidate() {
		p.addCandidateVote(pack.Head.CommitId == 1)
	}
}

func (p *protocol) on_CM_Append(conn *proto.ConnClient, pack *proto.PackData) {
	if !p.IsFollower() {
		logs.Warn("on_CM_Append 当前为:%s, 不再接收 CM_Append 数据", p.GetRoleName())
		return
	}

	if pack.Head.Term < p.Term {
		logs.Warn("on_CM_Append 收到小任期号的数据: %d, %d", pack.Head.Term, p.Term)
		return
	}

	logs.Printf("on_CM_Append: term = %d, dataIdx = %d", pack.Head.Term, pack.Head.DataIndex)

	var d types.AppendEntryData
	if types.Decode(pack.Data, &d) {
		logs.Printf("on_CM_Append: d = %#+v, myIndex = %d, commitIdx = %d", d, global.DataEntries.GetIndex(), global.DataEntries.GetCommitIdx())
		isAppendOK := false

		// 设置为正在追加数据的状态，这时候不学习
		p.isAppending = true
		p.appendingTime = time.Now()

		p.tempDatas[d.AppendIdx] = &d                         // 先放入临时列表，凑齐了再一起加过去
		if d.AppendIdx == global.DataEntries.GetCommitIdx() { // 数据收齐了，合并数据
			prevTerm := global.DataEntries.GetEntryItemTerm(d.AppendIdx - 1)
			isAppendOK = true
			logs.Printf("on_CM_Append: 数据收集齐了, %d -> %d", d.AppendIdx, pack.Head.DataIndex)
			for i := d.AppendIdx; i <= pack.Head.DataIndex; i++ {
				if v, ok := p.tempDatas[i]; ok {
					if prevTerm == v.PrevTerm {
						prevTerm = v.EntryItem.Term
						global.DataEntries.AppendEntries(&v.EntryItem)
						logs.Printf("on_CM_Append: append %#+v, idx = %d, count = %d", v.EntryItem, i, global.DataEntries.GetIndex())
						delete(p.tempDatas, i)
					} else {
						isAppendOK = false
						logs.Warn("on_CM_Append 数据ID不一致 %d %d", prevTerm, v.PrevTerm)
						break
					}
				} else {
					isAppendOK = false
					logs.Warn("on_CM_Append 合并数据失败，中间有空值")
					break
				}
			}
			if isAppendOK {
				p.isAppending = false // 结束  appending 状态
				dataIdx := global.DataEntries.GetIndex()
				lastCmtIdx := global.DataEntries.GetCommitIdx()
				if p.lastRecvCommitIdx > lastCmtIdx && p.lastRecvCommitIdx >= dataIdx {
					global.DataEntries.Commit(dataIdx)
					logs.Printf("on_CM_Append: 自动提交数据 %d -> %d, lastRecvCommitIdx = %d", lastCmtIdx, dataIdx, p.lastRecvCommitIdx)
				}
			}
		}

		resp := proto.NewDefPackWithData(proto.SM_Append, nil)
		resp.Head.CommitId = global.DataEntries.GetCommitIdx()
		if isAppendOK {
			resp.Head.DataIndex = pack.Head.DataIndex
		} else {
			resp.Head.DataIndex = d.AppendIdx - 1 // 继续要前一个数据
		}
		logs.Printf("on_CM_Append: isAppend = %v, GetIndex() = %d", isAppendOK, global.DataEntries.GetIndex())
		if err := conn.SendPackToQueue(resp); err != nil {
			logs.Warn("on_CM_Append error: %s ", err)
		}
	} else {
		logs.Printf("on_CM_Append: 解码数据失败")
	}
}

func (p *protocol) on_SM_Append(conn *proto.ConnClient, pack *proto.PackData) {
	if !p.IsLeader() {
		logs.Warn("on_SM_Append 当前为 %s, 不再接收 SM_Append 数据", p.GetRoleName())
		return
	}

	logs.Printf("on_SM_Append: term = %d, dataidx = %d", pack.Head.Term, pack.Head.DataIndex)

	if pack.Head.DataIndex == p.lastSendApppendIdx {
		p.appendRespCount++
	} else if pack.Head.DataIndex < p.lastSendApppendIdx {
		if data := global.DataEntries.GetEntryItem(pack.Head.DataIndex); data != nil {
			appendData := &types.AppendEntryData{EntryItem: *data, AppendIdx: pack.Head.DataIndex, PrevTerm: global.DataEntries.GetEntryItemTerm(pack.Head.DataIndex - 1)}
			logs.Printf("on_SM_Append: appendData = %#+v", appendData)
			if bits, ok := types.Encode(appendData); ok {
				resp := proto.NewDefPackWithData(proto.CM_Append, bits)
				resp.Head.Term = p.Term
				resp.Head.CommitId = global.DataEntries.GetCommitIdx()
				resp.Head.DataIndex = p.lastSendApppendIdx
				if err := conn.SendPackToQueue(resp); err != nil {
					logs.Warn("SM_Append SendPackToQueue error: %s", err)
				}
			}
		} else {
			logs.Warn("on_SM_Append GetEntryItem is nil, idx = %d", pack.Head.DataIndex)
		}
	} else {
		logs.Warn("on_SM_Append pack.Head.DataIndex error = %d, lastSendApppendIdx = %d", pack.Head.DataIndex, p.lastSendApppendIdx)
	}
}

func (p *protocol) on_CM_Commit(conn *proto.ConnClient, pack *proto.PackData) {
	if !p.IsFollower() {
		logs.Warn("on_CM_Commit 当前角色是 %s", p.GetRoleName())
		return
	}

	cmtIdx := global.DataEntries.GetCommitIdx()
	if pack.Head.CommitId > cmtIdx {
		term := global.DataEntries.GetEntryItemTerm(pack.Head.CommitId - 1)
		p.lastRecvCommitIdx = pack.Head.CommitId // 这里先记录下来，
		if pack.Head.Term == term {
			global.DataEntries.Commit(pack.Head.CommitId)
			logs.Warn("on_CM_Commit: 提交成功 %d -> %d", cmtIdx, pack.Head.CommitId)
			pack.Head.Cmd = proto.SM_Commit
			if err := conn.SendPackToQueue(pack); err != nil {
				logs.Printf("on_CM_Commit send pack error: %s", err)
			}
		} else {
			logs.Printf("on_CM_Commit: term 不一致 %d <> %d, pack.Head.CommitId = %d", pack.Head.Term, term, pack.Head.CommitId)
		}
	} else {
		logs.Warn("on_CM_Commit: 收到重复的 commit remote %d, local %d", pack.Head.CommitId, cmtIdx)
	}

}

func (p *protocol) on_SM_Commit(conn *proto.ConnClient, pack *proto.PackData) {
	if pack.Head.CommitId == p.lastSendCommitIdx {
		p.commitRespCount++
	}
}

func (p *protocol) on_CM_Learn(conn *proto.ConnClient, pack *proto.PackData) {
	var sendData *types.EntryItem
	if /* pack.Head.Term == 0 && */ pack.Head.DataIndex == 0 { // 第一个直接取
		sendData = global.DataEntries.GetEntryItem(pack.Head.DataIndex)
	} else { // 后面的要比较 term
		if data := global.DataEntries.GetEntryItem(pack.Head.DataIndex - 1); data != nil {
			if data.Term == pack.Head.Term { // 对得上，就直接发送数据了
				sendData = global.DataEntries.GetEntryItem(pack.Head.DataIndex)
			} else {
				logs.Printf("on_CM_Learn 任期对不上，学习失败: %d <> %d", data.Term, pack.Head.Term)
			}
		} else {
			logs.Printf("on_CM_Learn 获取不了数据: %d", pack.Head.DataIndex)
		}
	}

	var bits []byte
	if sendData != nil {
		var ok bool
		if bits, ok = types.Encode(sendData); !ok {
			logs.Printf("on_CM_Learn 编码失败")
		}
	} else {
		logs.Printf("on_CM_Learn 学习失败: term = %d, idx = %d", pack.Head.Term, pack.Head.DataIndex)
	}

	resp := proto.NewDefPackWithData(proto.SM_Learn, bits)
	resp.Head.Term = pack.Head.Term // 这两项数据原样返回
	resp.Head.DataIndex = pack.Head.DataIndex
	if err := conn.SendPackToQueue(resp); err != nil {
		logs.Printf("on_CM_Learn: 发送数据失败！")
	}
}

func (p *protocol) on_SM_Learn(conn *proto.ConnClient, pack *proto.PackData) {
	if len(pack.Data) > 0 { // 学习成功
		var data types.EntryItem
		if types.Decode(pack.Data, &data) {
			if err := global.DataEntries.UpdateEntries(pack.Head.DataIndex, &data); err != nil {
				logs.Printf("on_SM_Learn: updateEntries error %s", err)
				return
			} else {
				if err = global.DataEntries.Commit(pack.Head.DataIndex); err != nil {
					logs.Printf("on_SM_Learn 自动提交失败 commitIdx = %d, dataIndex = %d", pack.Head.DataIndex, global.DataEntries.GetIndex())
				} else {
					logs.Printf("on_SM_Learn: 自动提交成功 dataCount = %d, commitIdx = %d, data = %v", global.DataEntries.GetIndex(), pack.Head.DataIndex, data)
				}
			}

			if p.isAppending {
				logs.Printf("on_SM_Learn: 正在追加数据，终止学习")
			} else if pack.Head.DataIndex+1 < p.maxLearnIdx { // 继续学习下一条, 在非追加数据的状态下
				p.SendLearn(pack.Head.DataIndex+1, p.maxLearnIdx)
			}
		} else {
			logs.Printf("on_SM_Learn: decode 数据失败")
			return
		}
	} else {
		logs.Printf("on_SM_Learn: 学习失败 term = %d, idx = %d", pack.Head.Term, pack.Head.DataIndex)
		if pack.Head.DataIndex > 1 {
			p.SendLearn(pack.Head.DataIndex-1, 0) // 继续向前学习
		}
	}
}

func (p *protocol) addMsgHandles() {
	proto.AddMsgHandler(proto.CM_RequestVote, p.on_CM_RequestVote)
	proto.AddMsgHandler(proto.SM_RequestVote, p.on_SM_RequestVote)

	proto.AddMsgHandler(proto.CM_Append, p.on_CM_Append)
	proto.AddMsgHandler(proto.SM_Append, p.on_SM_Append)

	proto.AddMsgHandler(proto.CM_PING, p.on_CM_Ping)
	proto.AddMsgHandler(proto.SM_PONG, p.on_SM_Pong)

	proto.AddMsgHandler(proto.CM_Commit, p.on_CM_Commit)
	proto.AddMsgHandler(proto.SM_Commit, p.on_SM_Commit)

	proto.AddMsgHandler(proto.CM_Learn, p.on_CM_Learn)
	proto.AddMsgHandler(proto.SM_Learn, p.on_SM_Learn)
}

func newProto(own *roleState) *protocol {
	p := &protocol{roleState: own, pingPack: proto.NewDefPackWithData(proto.CM_PING, nil)}
	p.tempDatas = make(map[uint32]*types.AppendEntryData)
	p.addMsgHandles()
	return p
}
